﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using C5;
using Flower.Client;
using Flower.Logging;
using System.Transactions;
using System.Threading;
using Flower.Services;
using Flower.Services.Data;
using System.ServiceModel;
using Flower.Actors;
using Flower.Services.Clients;
using Flower.Directory.Util;

namespace Flower.StateService
{
	public sealed class StateService : IStateService, IDisposable
	{
        private class StateInfoTimeoutComparer : IComparer<StateInfo>
        {
            public int Compare(StateInfo x, StateInfo y)
            {
                return DateTime.Compare(x.ArrivedOn, y.ArrivedOn);
            }
        }

        private class DelayedAction
        {
            public DateTime When;
            public Action Action;
        }

        private class DelayedActionComparer : IComparer<DelayedAction>
        {
            public int Compare(DelayedAction x, DelayedAction y)
            {
                return DateTime.Compare(x.When, y.When);
            }
        }

        private static readonly Log Log = LogManager.CreateLog(typeof(StateService).Name);

        private const string PidResourcePrefix = "LATEST_STATE_";

        private static readonly TimeSpan DelayedActionCheckInterval = TimeSpan.FromSeconds(1);
        private static readonly TimeSpan StateTimeoutCheckInterval = TimeSpan.FromSeconds(1);
        private static readonly TimeSpan RestoreDelay = TimeSpan.FromSeconds(3);

	    private readonly StateServiceSettings _settings;

        private readonly Dictionary<string, StateInfo> _currentStates =
            new Dictionary<string, StateInfo>();
	    private readonly IStateStorage _storage;
	    private readonly IFlowerClient _flowerClient;
        private readonly IntervalHeap<StateInfo> _stateTimeouts =
            new IntervalHeap<StateInfo>(new StateInfoTimeoutComparer());
	    
        private readonly Thread _tasksCompletionWatch;
	    private readonly ManualResetEventSlim _stopTasksWatching = new ManualResetEventSlim(false);
        private readonly ConcurrentQueue<Task> _tasks = new ConcurrentQueue<Task>();

        private readonly Timer _stateTimeoutsTimer;

        private readonly Timer _retriesTimer;
        private readonly ArrayHeap<DelayedAction> _delayedActions =
            new ArrayHeap<DelayedAction>(new DelayedActionComparer());

        private volatile bool _isStopped;

        #region Startup

        public StateService(IFlowerClient flowerClient, IStateStorage storage, StateServiceSettings settings)
            : this(flowerClient, storage, settings, false)
        {
        }

        public StateService(IFlowerClient flowerClient, IStateStorage storage, StateServiceSettings settings, bool restoreWithoutDelay)
        {
            Log.Trace("Starting the state service.");

            try
            {
                _storage = storage;
                _flowerClient = flowerClient;
                _settings = settings;

                _tasksCompletionWatch = new Thread(TasksCompletionWatch);
                _tasksCompletionWatch.Start();

                _stateTimeoutsTimer =
                    new Timer
                    (
                        StateTimeoutManager,
                        null,
                        StateTimeoutCheckInterval,
                        StateTimeoutCheckInterval
                    );

                _retriesTimer = 
                    new Timer
                    (
                        CheckDelayedActions, 
                        null, 
                        DelayedActionCheckInterval,
                        DelayedActionCheckInterval
                    );

                if (restoreWithoutDelay)
                {
                    Restore();
                }
                else
                {
                    DelayAction(DateTime.Now + RestoreDelay, Restore);
                }
            }
            catch (Exception ex)
            {
                Log.Error("Startup failed.", ex);
                throw;
            }

            Log.Trace("The state service is started.");
        }

        #endregion

        #region IStateService implementation

        public Guid Save(string pid, Blob state, bool isKeyState)
        {
            if (pid == null) throw new ArgumentNullException("pid");
            if (state == null) throw new ArgumentNullException("state");
            if (_isStopped) throw new InvalidOperationException("Service is stopped.");

            var info = new StateInfo
            {
                Id = Guid.NewGuid(),
                Pid = pid,
                Data = state,
                ArrivedOn = DateTime.Now,
                Flags = (isKeyState) ? StateFlags.Key : StateFlags.None
            };

            DemandLock.Acquire
            (
                PidResourcePrefix + pid, 
                txn => 
                { 
                    if (txn.TransactionInformation.Status == TransactionStatus.Committed)
                    {
                        try
                        {
                            UpdateCurrentState(info);
                        }
                        catch (Exception ex)
                        {
                            Log.Error("Failed to update the current state.", ex);
                        }
                    }
                },
                null, 
                ex => Log.Error(string.Format("Failed to acquire resource {0}.", pid), ex)
            );

            _storage.Save(info);

            if (Transaction.Current == null)
            {
                UpdateCurrentState(info);
            }

            return info.Id;
        }

        public ProcessState GetLatestState(string pid)
	    {
            if (pid == null) throw new ArgumentNullException("pid");
            if (_isStopped) throw new InvalidOperationException("Service is stopped.");

            ProcessState result = null;

            TransactionHelper.Scope
            (
                IsolationLevel.ReadCommitted,
                TimeSpan.FromSeconds(30),
                () =>
                {
                    DemandLock.Acquire(PidResourcePrefix + pid, null, null, ex => Log.Error(string.Format("Failed to acquire resource {0}.", pid), ex));

                    lock (_currentStates)
                    {
                        StateInfo currentState;

                        if (_currentStates.TryGetValue(pid, out currentState))
                        {
                            result = new ProcessState
                            {
                                Id = currentState.Id,
                                Data = currentState.Data
                            };
                        }
                    }
                }
            );

            return result;
        }

	    public void MarkLatestStateAsKeyState(string pid)
	    {
            if (pid == null) throw new ArgumentNullException("pid");
            if (_isStopped) throw new InvalidOperationException("Service is stopped.");

            TransactionHelper.Scope
            (
                IsolationLevel.ReadCommitted,
                TimeSpan.FromSeconds(30),
                () =>
                {
                    DemandLock.Acquire(PidResourcePrefix + pid, null, null, ex => Log.Error(string.Format("Failed to acquire resource {0}.", pid), ex));

                    lock (_currentStates)
                    {
                        StateInfo currentState;

                        if (_currentStates.TryGetValue(pid, out currentState))
                        {
                            if (currentState.Flags.HasFlag(StateFlags.Key))
                            {
                                Log.Trace(() => string.Format("The latest state {0} of the process {1} is already a key state.", currentState.Id, pid));
                                return;
                            }

                            Log.Trace(() => string.Format("The latest state {0} of the process {1} is marked as a key state.", currentState.Id, pid));

                            currentState.Flags |= StateFlags.Key;
                            _storage.UpdateFlags(currentState.Id, currentState.Flags);

                            _tasks.Enqueue(Task.Factory.StartNew(() => PropagateStateToDirectory(currentState)));
                        }
                    }
                }
            );
        }

	    public void MarkLatestStateAsFinalState(string pid)
	    {
            if (pid == null) throw new ArgumentNullException("pid");
            if (_isStopped) throw new InvalidOperationException("Service is stopped.");

            TransactionHelper.Scope
            (
                IsolationLevel.ReadCommitted,
                TimeSpan.FromSeconds(30),
                () =>
                {
                    DemandLock.Acquire(PidResourcePrefix + pid, null, null, ex => Log.Error(string.Format("Failed to acquire resource {0}.", pid), ex));

                    lock (_currentStates)
                    {
                        StateInfo currentState;

                        if (_currentStates.TryGetValue(pid, out currentState))
                        {
                            Log.Trace(() => string.Format("The latest state {0} of the process {1} is marked as a final state.", currentState.Id, pid));

                            currentState.Flags |= StateFlags.Key | StateFlags.Final;
                            _storage.UpdateFlags(currentState.Id, currentState.Flags);

                            _tasks.Enqueue(Task.Factory.StartNew(() => PropagateStateToDirectory(currentState)));
                        }
                    }
                }
            );
        }

        #endregion

        #region Dispose

        public void Dispose()
	    {
            Log.Trace("Stopping state service.");

            try
            {
                _isStopped = true;

                _retriesTimer.Dispose();
                _stateTimeoutsTimer.Dispose();

                _stopTasksWatching.Set();
                _tasksCompletionWatch.Join();

                var disposableStorage = _storage as IDisposable;
                if (disposableStorage != null)
                {
                    disposableStorage.Dispose();
                }
            }
            catch (Exception ex)
            {
                Log.Error("Shutdown failed.", ex);
                throw;
            }

            Log.Trace("State service is stopped.");
	    }

        #endregion

        #region Restore

        private void Restore()
        {
            Log.Trace("Restoring current states.");

            foreach (StateInfo state in _storage.GetAllStates())
            {
                UpdateCurrentState(state);
            }
        }

        #endregion

        #region Task completion watch

        private void TasksCompletionWatch()
        {
            Log.Trace("Entering task completion watching thread.");

            bool run;

            do
            {
                run = !_stopTasksWatching.Wait(TimeSpan.FromSeconds(1));

                Task task;

                while (_tasks.TryDequeue(out task))
                {
                    try
                    {
                        task.Wait();
                    }
                    catch(Exception ex)
                    {
                        Log.Error("Task failed.", ex);
                    }
                }

            } while (run);

            Log.Trace("Leaving task completion watching thread.");
        }

        #endregion

        #region State timeout manager

        private void StateTimeoutManager(object arg)
        {
            Log.Trace("Checking state timeouts.");

            if (!_stateTimeouts.IsEmpty)
            {
                lock (_currentStates)
                {
                    StateInfo oldestState;
                    DateTime limit = DateTime.Now - _settings.StateTimeout;

                    while (!_stateTimeouts.IsEmpty && (oldestState = _stateTimeouts.FindMin()).ArrivedOn < limit)
                    {
                        _stateTimeouts.DeleteMin();
                        oldestState.Handle = null;
                        oldestState.Flags |= StateFlags.Key | StateFlags.Final;

                        if (!oldestState.Flags.HasFlag(StateFlags.Key))
                        {
                            var state = oldestState;

                            Log.Trace(() => string.Format("State {0} of the process {1} is timed out and will be propagated to the directory.", state.Id, state.Pid));

                            _tasks.Enqueue(Task.Factory.StartNew(() => PropagateStateToDirectory(state)));
                        }
                    }
                }
            }
        }

        #endregion

        #region Current state management

        private void RegisterState(StateInfo state, bool setTimeout)
        {
            Log.Trace(() => string.Format("Registering state {0}{1}.", state.Id, (state.Flags.HasFlag(StateFlags.Key)) ? " as key state" : string.Empty));

            _currentStates.Add(state.Pid, state);

            if (setTimeout)
            {
                IPriorityQueueHandle<StateInfo> handle = null;
                _stateTimeouts.Add(ref handle, state);
                state.Handle = handle;
            }
        }

        private void UnregisterState(StateInfo state)
        {
            Log.Trace(() => string.Format("Unregistering state {0}.", state.Id));

            _currentStates.Remove(state.Pid);

            if (state.Handle != null)
            {
                _stateTimeouts.Delete(state.Handle);
                state.Handle = null;
            }
        }

        private void UpdateCurrentState(StateInfo state)
        {
            Log.Trace(() => string.Format("Updating state of the process {0} {1}.", state.Pid, state.Id));

            lock (_currentStates)
            {
                StateInfo prevState;

                if (_currentStates.TryGetValue(state.Pid, out prevState))
                {
                    Log.Trace(() => string.Format("Previous state {0} of the process {1} found and will be deleted.", prevState.Id, state.Pid));

                    UnregisterState(prevState);

                    _tasks.Enqueue
                    (
                        Task.Factory.StartNew
                        (
                            () =>
                            {
                                try
                                {
                                    Log.Trace(() => string.Format("Deleting state {0}.", prevState.Id));
                                    _storage.Delete(prevState.Id);
                                }
                                catch (Exception ex)
                                {
                                    Log.Error("Failed to delete a state.", ex);
                                }
                            }
                        )
                    );
                }

                RegisterState(state, !state.Flags.HasFlag(StateFlags.Key));

                if (state.Flags.HasFlag(StateFlags.Key))
                {
                    _tasks.Enqueue(Task.Factory.StartNew(() => PropagateStateToDirectory(state)));
                }
            }
        }

        private void PropagateStateToDirectory(StateInfo state)
        {
            Log.Trace(() => string.Format("Propagating state {0} to the directory.", state.Id));

            try
            {
                TransactionHelper.Scope
                (
                    IsolationLevel.ReadCommitted, 
                    _settings.DirectoryPropagationTransactionTimeout,
                    () =>
                    {
                        try
                        {
                            _flowerClient
                                .CreateScript()
                                .SetText("@SaveState")
                                .AddParameter("pid", state.Pid)
                                .AddParameter
                                (
                                    "state",
                                    new State
                                    {
                                        Name = state.Id.ToString(),
                                        Data = state.Data
                                    }
                                )
                                .Execute();
                        }
                        catch (FaultException<DirectoryFault> ex)
                        {
                            if
                            (
                                //The process has been deleted from the directory.
                                ex.Detail.Code != DirectoryFaultCode.EntryNotFound &&
                                //The state has already been saved.
                                ex.Detail.Code != DirectoryFaultCode.NameCollision
                            )
                            {
                                throw;
                            }
                        }

                        if (state.Flags.HasFlag(StateFlags.Final))
                        {
                            Log.Trace(() => string.Format("Deleting state {0} after propagation to the directory.", state.Id));
                            _storage.Delete(state.Id);

                            Transaction.Current.TransactionCompleted +=
                                (sender, args) =>
                                {
                                    if (args.Transaction.TransactionInformation.Status == TransactionStatus.Committed)
                                    {
                                        try
                                        {
                                            lock (_currentStates)
                                            {
                                                UnregisterState(state);
                                            }
                                        }
                                        catch (Exception ex)
                                        {
                                            Log.Error("Failed to unregister state on transaction commit.", ex);
                                        }
                                    }
                                };
                        }
                    }
                );
            }
            catch (WcfClientException ex)
            {
                if (ex.InnerException is FaultException)
                {
                    Log.Error(string.Format("Failed to propagate the state {0} to the directory.", state.Id), ex);
                }
                else
                {
                    if (!_isStopped)
                    {
                        Log.Warning
                        (
                            string.Format
                            (
                                "Failed to propagate the state {0} to the directory due to communication error '{1}'. Will be retried at {2}.",
                                state.Id,
                                ex.Message,
                                DateTime.Now + _settings.DirectoryPropagationRepeatOnFailureInterval
                            )
                        );

                        DelayAction
                        (
                            DateTime.Now + _settings.DirectoryPropagationRepeatOnFailureInterval,
                            () => PropagateStateToDirectory(state)
                        );
                    }
                }
            }
            catch (Exception ex)
            {
                Log.Error(string.Format("Failed to propagate the state {0} to the directory.", state.Id), ex);
            }
        }

        #endregion

        #region Retries management

        private void DelayAction(DateTime due, Action action)
        {
            lock (_delayedActions)
            {
                _delayedActions.Add(new DelayedAction { When = due, Action = action });
            }
        }

        private void CheckDelayedActions(object arg)
        {
            Log.Trace("Checking delayed actions.");

            if (_delayedActions.Count > 0)
            {
                lock (_delayedActions)
                {
                    var now = DateTime.Now;

                    DelayedAction closestAction;

                    while (_delayedActions.Count > 0 && (closestAction = _delayedActions.Peek()).When < now)
                    {
                        _tasks.Enqueue(Task.Factory.StartNew(closestAction.Action));
                        _delayedActions.Pop();
                    }
                }
            }
        }

        #endregion
    }
}
